package com.burges.net.tableAPIAndSQL.userDefinedDataSource.tableFactory

import java.util

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.descriptors.ConnectorDescriptor

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/25 
  * 描述      在Table&SQL API中使用TableFactory
  */
class MySocketConnector(host: String,port:String) extends ConnectorDescriptor("dev-system", 1, false) {

	override protected def toConnectorProperties(): Map[String, String] = {
		val properties = new util.HashMap[String, String]()
		properties.put("connector.host", host)
		properties.put("connector.port", port)
		properties
	}
}

object MySocketConnector{

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tStreamEnv = TableEnvironment.getTableEnvironment(environment)

		val mySocketTable: Unit = tStreamEnv.connect(new MySocketConnector("localhost", "10000"))
				.inAppendMode()
				.registerTableSource("mySocketTable")
	}
}
